#ifndef _RPC_DISPATCHER_H_
#define _RPC_DISPATCHER_H_

#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>
#include <google/protobuf/service.h>
#include <map>
#include <memory>
#include <shared_mutex>
#include "fastrpc/rpc/coder/fastpb_protocol.hpp"
#include "fastrpc/rpc/common/run_time.hpp"
#include "fastrpc/rpc/fastrpc/fastrpc_controller.hpp"
namespace fastrpc::rpc::detail {

class RpcDispatcher {
public:
    inline static RpcDispatcher *g_rpc_dispatcher = NULL;
    static RpcDispatcher        *GetRpcDispatcher() {
        if (g_rpc_dispatcher == NULL) {
            g_rpc_dispatcher = new RpcDispatcher();
        }
        return g_rpc_dispatcher;
    }

public:
    using service_s_ptr = std::shared_ptr<google::protobuf::Service>;

    void dispatch(FastPBProtocol::s_ptr            request,
                  FastPBProtocol::s_ptr            response,
                  fastrpc::socket::net::TcpStream &connection,
                  google::protobuf::Closure       *done = NULL) {
        std::shared_ptr<FastPBProtocol> req_protocol
            = std::dynamic_pointer_cast<FastPBProtocol>(request);
        std::shared_ptr<FastPBProtocol> rsp_protocol
            = std::dynamic_pointer_cast<FastPBProtocol>(response);

        // 初始化响应头
        rsp_protocol->m_msg_id = req_protocol->m_msg_id;
        rsp_protocol->m_method_name = req_protocol->m_method_name;

        // 解析服务和方法名称
        std::string service_name, method_name;
        if (!parseServiceFullName(req_protocol->m_method_name, service_name, method_name)) {
            setFastPBError(rsp_protocol,
                           fastrpc::Error::ERROR_PARSE_SERVICE_NAME,
                           "Invalid service format");
            return;
        }

        // 服务查找
        service_s_ptr service;
        auto          name = m_service_map.find(service_name);
        if (name == m_service_map.end()) {
            setFastPBError(rsp_protocol,
                           fastrpc::Error::ERROR_SERVICE_NOT_FOUND,
                           "Service not registered");
            return;
        }
        service = name->second;

        // 方法查找
        const google::protobuf::MethodDescriptor *method;
        std::string                               cache_key = service_name + "." + method_name;
        // console.info("cache_key:{}", cache_key);
        auto it = method_cache_.find(cache_key);
        if (it == method_cache_.end()) {
            setFastPBError(rsp_protocol,
                           fastrpc::Error::ERROR_METHOD_NOT_FOUND,
                           "Method not registered");
            return;
        }
        method = it->second;

        // 请求反序列化
        std::unique_ptr<google::protobuf::Message> req_msg(
            service->GetRequestPrototype(method).New());
        if (!req_msg->ParseFromString(req_protocol->m_pb_data)) {
            setFastPBError(rsp_protocol,
                           fastrpc::Error::ERROR_FAILED_DESERIALIZE,
                           "Deserialization failed");
            return;
        }

        // 准备响应
        std::unique_ptr<google::protobuf::Message> rsp_msg(
            service->GetResponsePrototype(method).New());

        // 执行RPC调用
        FastRpcController controller;
        controller.SetLocalAddr(connection.local_addr().value());
        controller.SetPeerAddr(connection.peer_addr().value());
        controller.SetMsgId(req_protocol->m_msg_id);

        util::RunTime::GetRunTime()->m_msg_id = req_protocol->m_msg_id;
        util::RunTime::GetRunTime()->m_method_name = method_name;

        service->CallMethod(method, &controller, req_msg.get(), rsp_msg.get(), done);

        // 序列化响应
        if (!rsp_msg->SerializeToString(&rsp_protocol->m_pb_data)) {
            setFastPBError(rsp_protocol,
                           fastrpc::Error::ERROR_FAILED_SERIALIZE,
                           "Serialization failed");

            return;
        }

        rsp_protocol->m_err_code = 0;
    }

    void registerService(service_s_ptr service) {
        std::string service_name = service->GetDescriptor()->full_name();
        m_service_map[service_name] = service;
        auto descriptor = service->GetDescriptor();
        for (int i = 0; i < descriptor->method_count(); ++i) {
            auto        method = descriptor->method(i);
            std::string key = service_name + "." + method->name();
            // console.info("key:{}", key);
            {
                method_cache_[key] = method;
            }
        }
    }

    void setFastPBError(std::shared_ptr<FastPBProtocol> msg,
                        int32_t                         err_code,
                        const std::string               err_info) {
        msg->m_err_code = err_code;
        msg->m_err_info = err_info;
        msg->m_err_info_len = err_info.length();
    }

    std::map<std::string, service_s_ptr> get_service_map() {
        return m_service_map;
    }

private:
    bool parseServiceFullName(const std::string &full_name,
                              std::string       &service_name,
                              std::string       &method_name) {
        if (full_name.empty()) {
            LOG_ERROR("full name empty");
            return false;
        }
        size_t i = full_name.find_first_of(".");
        if (i == full_name.npos) {
            LOG_ERROR("can not found . in full name [%s]", full_name);
            return false;
        }
        service_name = full_name.substr(0, i);
        method_name = full_name.substr(i + 1, full_name.length() - i - 1);

        LOG_INFO("parse service_name[%s] ans method_name[%s] from full name[%s]",
                 service_name.c_str(),
                 method_name.c_str(),
                 full_name.c_str());
        return true;
    }

private:
    std::map<std::string, service_s_ptr>                                        m_service_map;
    std::unordered_map<std::string, const google::protobuf::MethodDescriptor *> method_cache_;
};

}  // namespace fastrpc::rpc::detail

#endif
